package com.smartstate.iot.mqtt.callback;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.smartstate.iot.config.SpringBeanUtils;
import com.smartstate.iot.entity.TopicConfig;
import com.smartstate.iot.entity.TopicLog;
import com.smartstate.iot.mapper.TopicLogMapper;
import com.smartstate.iot.mapper.TopicMapper;
import com.smartstate.iot.mqtt.client.MyMQTTClient;
import com.smartstate.iot.service.*;
import com.smartstate.iot.service.impl.DataDistributionServiceImpl;
import com.smartstate.iot.tool.TopicTool;
import com.smartstate.iot.vo.TopicUpVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.util.buf.Utf8Decoder;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MyMQTTCallback implements MqttCallback {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTCallback.class);

    private MyMQTTClient myMQTTClient;


    private static final ThreadPoolExecutor threadpool = new ThreadPoolExecutor(100, 200,
            10L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

    public MyMQTTCallback(MyMQTTClient myMQTTClient) {
        this.myMQTTClient = myMQTTClient;
    }

    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        try {
            myMQTTClient.connect();
            myMQTTClient.subscribe("SH/#", 1);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param topic
     * @param message
     * @throws Exception subscribe后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String msg = new String(message.getPayload(), "UTF-8");
        //  采集盒子
        if (topic.contains("ALL")) {
            if (topic.endsWith("HEART")) {   //  采集盒子心跳
//                Map<String, Object> map = JSON.parseObject(msg, Map.class);
                TopicUpVo topicUpVo = JSONObject.parseObject(msg, TopicUpVo.class);
                TopicTool bean = SpringBeanUtils.getBean(TopicTool.class);
                //  查询数据库，如果是首次心跳，下发newTopic
                Boolean flag = bean.isFirstHeartbeat(topicUpVo);
                if (flag) {
                    //  设置newTopic并返回给盒子
                    for (TopicUpVo.Content content1 : topicUpVo.getContent()) {
                        String deviceId = content1.getDeviceId();
                        content1.setNewTopic("SH/HK/NJ/UP/ALL/" + deviceId);
                    }
                    String newConfig = JSON.toJSONString(topicUpVo);
                    myMQTTClient.publish(newConfig, "SH/HK/NJ/CONFIG", 2, false);
                    //  将首次心跳中携带的配置参数发送给中台
                    List<TopicUpVo.Content> contents = topicUpVo.getContent();
                    for (TopicUpVo.Content content : contents) {
                        List<TopicUpVo.ConfigDetails> config = content.getConfig();
                        myMQTTClient.publish(JSON.toJSONString(config), "BOX/CONFIG/UP/ZT/FIRST");
                    }
                }
                //  储存心跳
//                TopicService topicService = SpringBeanUtils.getBean(TopicService.class);
//                topicService.saveByDeviceId(topicUpVo);
                //  将心跳信息储存
                Boolean Boolean = bean.saveTopicConfig(topicUpVo);
            } else if (topic.endsWith("CONFIG")) {
                //  iot下发给盒子，消息过滤，不做处理
            } else if (topic.endsWith("CONFIGS")) {   //  中台下发盒子报警配置
                threadpool.execute(new TransmitThreadService("SH/HK/NJ/CONFIG", msg));
            } else {     //  盒子采集数据
                //  盒子采集的信息储存在redis
                threadpool.execute(new ThreadService(topic, msg));
                Map map = JSON.parseObject(msg, Map.class);
                List<TopicUpVo.ConfigDetails> config = (List<TopicUpVo.ConfigDetails>) map.get("config");
                String newMsg = config.toString();
//                List<TopicUpVo.ConfigDetails> config = (List<TopicUpVo.ConfigDetails>) map.get("config");
//                String objects = JSON.toJSONString(config);
//                String configured = (String) map.get("configured");
//                JSONObject jsonObject = new JSONObject();
//                jsonObject.put("configured",configured);
//                jsonObject.put("config",objects);
//                String newMsg = JSON.toJSONString(jsonObject);
                myMQTTClient.publish(newMsg,"BOX/CONFIG/UP/ZT");
            }
        } else if (topic.contains("PLC")) {   //采集PLC
            if (topic.endsWith("HEART")) {   //  采集盒子心跳
                Map<String, Object> map = JSON.parseObject(msg, Map.class);
                TopicUpVo beatVO = new TopicUpVo();
                beatVO.setConfig(map.get("config").toString());
                beatVO.setTimesTamp(map.get("timesTamp").toString());
                ArrayList<TopicUpVo.Content> contents = new ArrayList<>();
                TopicUpVo.Content content1 = new TopicUpVo.Content();
                Map<String,Object> content = (Map) map.get("content");
                content1.setDeviceId(content.get("boxId").toString());
                content1.setDeviceName(content.get("boxName").toString());
                content1.setDeviceType(content.get("boxType").toString());
                content1.setIp(content.get("ip").toString());
                content1.setPort(content.get("port").toString());
                String cpuTemp = content.get("cpu").toString();
                String cpu = cpuTemp.substring(0,cpuTemp.length()-1);
                String memoryTemp = content.get("memory").toString();
                String memory = memoryTemp.substring(0,memoryTemp.length()-1);
                content1.setCpu(cpu);
                content1.setMomery(memory);
                content1.setNewTopic(content.get("newTopic").toString());
                contents.add(content1);
                beatVO.setContent(contents);
                TopicTool bean = SpringBeanUtils.getBean(TopicTool.class);
                //  将心跳信息储存
                Boolean Boolean = bean.saveTopicConfig(beatVO);
            } else if (topic.endsWith("CONFIG")) {
                //  iot下发给盒子，消息过滤，不做处理
            } else if (topic.endsWith("CONFIGS")) {   //  中台下发盒子报警配置
                threadpool.execute(new PointsAwardedThreadService("SH/HK/NJ/PLC/CONFIG", msg));
            } else {     //  盒子采集数据
                threadpool.execute(new PlcThreadService(topic, msg));
            }
        }
        log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(message.getPayload(), "UTF-8"));
    }

    /**
     * Json格式转译
     *
     * @param jsonObj
     * @return
     */
    public Map<String, Object> analysis(JSONObject jsonObj) {
        JSONArray readResults = jsonObj.getJSONArray("values");
        Map<String, Object> map = new HashMap<>();
        for (int i = 0; i < readResults.size(); i++) {
            JSONObject jsonObject = readResults.getJSONObject(i);
            String[] split = jsonObject.getString("id").split("\\.");
            String id = split[split.length - 1];
            Object value = 0;
            if (Boolean.parseBoolean(jsonObject.get("q").toString())) {
                value = jsonObject.get("v");
            }
            map.put(id, value);
        }
        return map;
    }

    /**
     * 消息到达后
     * subscribe后，执行的回调函数
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */

    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//        LOGGER.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }
}
